
#include "StorexEngine.h"


StorexEngine::StorexEngine(ConfigHandler* parConfigHandler, OutputListener* outputListener):
	resultListener(*this),
	outputListener(outputListener)
{

	memory = new MemoryWrapper(parConfigHandler->getStorageType(), parConfigHandler->getIndexingType());

	if (parConfigHandler->getSubManagerType() == SIENAFF){
		subsManagerType = SIENAFF;
		subsManager = new SienaInterface();

		if (parConfigHandler->getMemoryTableType() == PRIORITY){
			memoryTable = new PriorityMemoryTable(parConfigHandler->getNumSlot(), memory);
		} else
		if (parConfigHandler->getMemoryTableType() == ALLOCATION){
			memoryTable = new AllocationMemoryTable(parConfigHandler->getNumSlot(), memory);
		}

		if (parConfigHandler->getMemManagerExecutionFlow() ==  SERIAL){
			if (parConfigHandler->getMemoryTableType() == PRIORITY){
				memoryManager = new SerialMemoryManager2(memoryTable, parConfigHandler->getSubsThreshold(), parConfigHandler->getNumInterval(), parConfigHandler->getLastThreshold());
			} else
			if (parConfigHandler->getMemoryTableType() == ALLOCATION){
				memoryManager = new SerialMemoryManager(memoryTable, parConfigHandler->getSubsThreshold(), parConfigHandler->getNumInterval(), parConfigHandler->getLastThreshold());
			}

		} else
		if (parConfigHandler->getMemManagerExecutionFlow() == PARALLEL){
			memoryManager = new ParallelMemoryManager(memoryTable, parConfigHandler->getSubsThreshold(), parConfigHandler->getNumInterval(), parConfigHandler->getLastThreshold());
		}
	} else
	if (parConfigHandler->getSubManagerType() == OCM){
		subsManagerType = OCM;
		subsManager = new OMPInterface();

		if (parConfigHandler->getMemoryTableType() == FIFO){
			ocmMemoryTable = new FIFOMemoryTable(parConfigHandler->getNumSlot(), memory);
		} else
		if (parConfigHandler->getMemoryTableType() == EVENT){
			float* point = new float[20];
			point[0] = 0.000003;
			point[1] = 0.000006;
			point[2] = 0.000009;
			point[3] = 0.000012;
			point[4] = 0.000018;
			point[5] = 0.000021;
			point[6] = 0.000024;
			point[7] = 0.000027;
			point[8] = 0.000030;
			point[9] = 0.000033;
			point[10] = 0.000036;
			point[11] = 0.000039;
			point[12] = 0.000042;
			point[13] = 0.000045;
			point[14] = 0.000048;
			point[15] = 0.000051;
			point[16] = 0.000054;
			point[17] = 0.000057;
			point[18] = 0.000060;
			point[19] = 0.000063;
			ocmMemoryTable = new EventMemoryTable(parConfigHandler->getNumSlot(), memory, point, 20, parConfigHandler->getLastThreshold());
		}
	}

	trexEngine = new TRexEngine(parConfigHandler->getNumProc());
	trexEngine->finalize();
	trexEngine->addResultListener(&resultListener);

	timer = new Timer();
}


StorexEngine::~StorexEngine(){
	delete memory;
	delete memoryTable;
	delete ocmMemoryTable;
	delete memoryManager;
	delete trexEngine;
	delete timer;
}


void StorexEngine::processRulePkt(RulePkt *pkt){

	ruleSet.insert(pkt);
	if (ruleSet.size() == 10){
		for (set<RulePkt*>::iterator it = ruleSet.begin(); it!=ruleSet.end(); it++){
			RulePkt* pkt2 = *it;
			trexEngine->processRulePkt(pkt2);
			cout << "Inserted rule into engine" << endl;
		}
		trexEngine->finalize();
		trexEngine->addResultListener(&resultListener);
	}
}


void StorexEngine::processSubPkt(Inbox* inbox){
	double startTime;
	double endTime;
	timer->start();
	cout << "Subscription manager: start processing subscription..." << endl;
	startTime=  timer->getElapsedTimeInMilliSec();
	subsManager->installSubscription(inbox);
	endTime=  timer->getElapsedTimeInMilliSec();
	cout << "Subscription manager: subscription processed in " << (endTime - startTime) << " ms." << endl;
	cout << "Memory manager: start processing subscription..." << endl;
	std::set<int> clients;
	clients.insert(inbox->getClient());
	std::set<PubPkt*> pkts;
	Outbox outbox = Outbox(clients, pkts);
	if (subsManagerType == SIENAFF) {
		startTime=  timer->getElapsedTimeInMilliSec();
		memoryManager->queryEventSet(inbox->getSubPkt(), outbox);
		endTime=  timer->getElapsedTimeInMilliSec();
		cout << "Memory manager: subscription processed .in " << (endTime - startTime) << " ms." << endl;
		cout << "Memory manager: start collecting subscription..." << endl;
		startTime=  timer->getElapsedTimeInMilliSec();
		memoryManager->addSubscription(inbox->getSubPkt());
		endTime=  timer->getElapsedTimeInMilliSec();
		cout << "Memory manager: subscription collected in " << (endTime - startTime) << " ms." << endl;
	} else if (subsManagerType == OCM) {
		startTime=  timer->getElapsedTimeInMilliSec();
		ocmMemoryTable->computeExtraction(inbox->getSubPkt(), outbox.pkts);
		endTime=  timer->getElapsedTimeInMilliSec();
		cout << "Memory manager: subscription processed in " << (endTime - startTime) << " ms." << endl;
	}
	timer->stop();
	cout << "Subscription completely processed in " << timer->getElapsedTimeInMilliSec() << " ms." << endl;
	outputListener->handleOutput(outbox, timer->getElapsedTimeInMilliSec());
}


void StorexEngine::processUnSubPkt(Inbox* inbox){
	subsManager->deleteSubscription(inbox);
}


void StorexEngine::processPubPkt(PubPkt *pkt){
	double startTime;
	double endTime;
	timer->start();
	cout << "TRex engine: start processing message..." << endl;
	TimeMs eventTimestamp = pkt->getTimeStamp();
	pkt->incRefCount();
	startTime=  timer->getElapsedTimeInMilliSec();
	trexEngine->processPubPkt(pkt);
	endTime=  timer->getElapsedTimeInMilliSec();
	pkt->setTime(eventTimestamp);
	cout << "TRex engine: message processed in " << (endTime - startTime) << " ms." << endl;
	//ha messo i risultati nel listener
	std::set<PubPkt*> pktSet;
	std::set<int> clientSet;
	Outbox outbox(clientSet, pktSet);
	if (subsManagerType == SIENAFF) {
		cout << "Subscription manager: start processing message..." << endl;
		startTime=  timer->getElapsedTimeInMilliSec();
		subsManager->processEvent(pkt, outbox);
		endTime=  timer->getElapsedTimeInMilliSec();
		cout << "Subscription manager: message processed in " << (endTime - startTime) << " ms." << endl;
		cout << "Memory manager: start storing message..." << endl;
		startTime=  timer->getElapsedTimeInMilliSec();
		memoryManager->addEvent(pkt);
		endTime=  timer->getElapsedTimeInMilliSec();
		cout << "Memory manager: message stored in " << (endTime - startTime) << " ms." << endl;
	} else if (subsManagerType == OCM){
		cout << "Subscription manager: start processing message..." << endl;
		startTime=  timer->getElapsedTimeInMilliSec();
		subsManager->processEvent(pkt, outbox);
		endTime=  timer->getElapsedTimeInMilliSec();
		cout << "Subscription manager: message processed in " << (endTime - startTime) << " ms." << endl;
		cout << "Memory manager: start storing message..." << endl;
		startTime=  timer->getElapsedTimeInMilliSec();
		ocmMemoryTable->computeInsertion(pkt, outbox.getNs(), outbox.getWs(), outbox.getLastPercentage());
		endTime=  timer->getElapsedTimeInMilliSec();
		cout << "Memory manager: message stored in " << (endTime - startTime) << " ms." << endl;
	}
	timer->stop();
	cout << "Message completely processed in " << timer->getElapsedTimeInMilliSec() << " ms." << endl;


	outbox.pkts.clear();
	outputListener->handleOutput(outbox, timer->getElapsedTimeInMilliSec());


	std::set<PubPkt *> resultSet = resultListener.getEventsSet();
	std::queue<PubPkt *> pktQueue;
	for (std::set<PubPkt *>::iterator it = resultSet.begin(); it != resultSet.end(); it++){
		PubPkt* currPkt = *it;
		timer->start();
		std::set<PubPkt*> pktSet;
		std::set<int> clientSet;
		Outbox outbox(clientSet, pktSet);

		if (subsManagerType == SIENAFF) {
			cout << "Subscription manager: start processing message..." << endl;
			startTime=  timer->getElapsedTimeInMilliSec();
			subsManager->processEvent(currPkt, outbox);
			endTime=  timer->getElapsedTimeInMilliSec();
			cout << "Subscription manager: message processed in " << (endTime - startTime) << " ms." << endl;
			cout << "Memory manager: start storing message..." << endl;
			startTime=  timer->getElapsedTimeInMilliSec();
			memoryManager->addEvent(currPkt);
			endTime=  timer->getElapsedTimeInMilliSec();
			cout << "Memory manager: message stored in " << (endTime - startTime) << " ms." << endl;
		} else if (subsManagerType == OCM){
			cout << "Subscription manager: start processing message..." << endl;
			startTime=  timer->getElapsedTimeInMilliSec();
			subsManager->processEvent(currPkt, outbox);
			endTime=  timer->getElapsedTimeInMilliSec();
			cout << "Subscription manager: message processed in " << (endTime - startTime) << " ms." << endl;
			cout << "Memory manager: start storing message..." << endl;
			startTime=  timer->getElapsedTimeInMilliSec();
			ocmMemoryTable->computeInsertion(currPkt, outbox.getNs(), outbox.getWs(), outbox.getLastPercentage());
			endTime=  timer->getElapsedTimeInMilliSec();
			cout << "Memory manager: message stored in " << (endTime - startTime) << " ms." << endl;
		}

		timer->stop();
		cout << "Message completely processed in " << timer->getElapsedTimeInMilliSec() << " ms." << endl;

		outbox.pkts.clear();
		outputListener->handleOutput(outbox, timer->getElapsedTimeInMilliSec());

		if (recursiveTypes.find(currPkt->getEventType()) != recursiveTypes.end()){
			pktQueue.push(currPkt);
		}
	}
/*	while (!pktQueue.empty()){
		PubPkt* currRecPkt = pktQueue.front();
		pktQueue.pop();
		trexEngine->processPubPkt(currRecPkt);
		resultSet = resultListener.getEventsSet();
		for (std::set<PubPkt *>::iterator it = resultSet.begin(); it != resultSet.end(); it++){
			PubPkt* currPkt = *it;
			std::set<PubPkt*> pktSet;
			std::set<int> clientSet;
			Outbox outbox(clientSet, pktSet);
			subsManager->processEvent(currPkt, outbox);
			memoryManager->addEvent(currPkt);
			outputListener->handleOutput(outbox, 0);
			if (recursiveTypes.find(currPkt->getEventType()) != recursiveTypes.end()){
				pktQueue.push(currPkt);
			}
		}
	}
*/
}


void StorexEngine::TRexResultListener::handleResult(set<PubPkt *> &genPkts, double procTime){
	eventsSet.clear();
	for (std::set<PubPkt*>::iterator it = genPkts.begin(); it != genPkts.end(); it++){
		(*it)->incRefCount();
		eventsSet.insert(*it);
		cout << "EVENT SET RIEMPITO!!!" << endl;
	}

}







